package cn.qiujing.java.study.demo.kafka.manage;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class KafkaConsumer {



    private final String topics = "topic1";

    private final String groupId = "test_consumer_group";

    /**
     * 消费监听，指定消费的主题，指定消费组
     *
     * @param record
     */
    //@KafkaListener(topics = {topics}, groupId = groupId)
    public void onMessage1(ConsumerRecord<?, ?> record) {
        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.println("简单消费1：" + record.topic() + "-" + record.partition() + "-" + record.value());
    }

    /**
     * 消费监听，指定消费的主题，指定消费组
     *
     * @param record
     */
    //@KafkaListener(topics = {topics}, groupId = groupId)
    public void onMessage2(ConsumerRecord<?, ?> record) {
        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.println("简单消费2：" + record.topic() + "-" + record.partition() + "-" + record.value());
    }

    @KafkaListener(topics = {topics}, containerFactory = "simpleFactory")
    public void onMessage3(ConsumerRecord<?, ?> record, Acknowledgment ack) {
        // 消费的哪个topic、partition的消息,打印出消息内容
        ack.acknowledge();
        System.out.println("手动ack消费：" + record.topic() + "-" + record.partition() + "-" + record.value());
    }


}
